use std::{path::PathBuf, pin::Pin};

use crate::{
    app::{
        DfsGrpcService, FileProcessResult, MetadataDownloadRequest, P2pCommand, Processor,
        store::{self, PendingDownloadRecord, PublicPublishedFile},
    },
    pb::{
        DownloadFileRequest, DownloadFileResponse, GetPublicFilesRequest, GetPublicFilesResponse,
        PublishFileRequest, PublishFileResponse,
        dfs_server::{Dfs, DfsServer},
    },
};
use tokio::sync::{mpsc, oneshot};
use tokio_stream::{Stream, wrappers::ReceiverStream};
use tokio_util::sync::CancellationToken;
use tonic::transport::Server;
use tonic::{Request, Response, Status};
use tracing::{error, info};

use crate::app::Service;
use crate::error::ServerError;

const LOG_TARGET: &str = "app::grpc::dfs";

pub type ServiceResult<T> = Result<Response<T>, Status>;
pub type GetPublicFilesStream =
    Pin<Box<dyn Stream<Item = Result<GetPublicFilesResponse, Status>> + Send>>;

#[tonic::async_trait]
impl<FileStore> Dfs for DfsGrpcService<FileStore>
where
    FileStore: store::Store + Send + Sync + 'static,
{
    type GetPublicFilesStream = GetPublicFilesStream;

    async fn publish_file(
        &self,
        request: Request<PublishFileRequest>,
    ) -> ServiceResult<PublishFileResponse> {
        let request = request.into_inner();
        info!(target: LOG_TARGET, "We got a new publish file request: {:?}", request);

        let file_processor = Processor::new();
        let file_process_result = file_processor.process_file(&request).await?;

        self.file_publish_sender
            .send(file_process_result)
            .await
            .map_err(|e| Status::internal(format!("cannot send file process result: {}", e)))?;

        Ok(Response::new(PublishFileResponse {
            success: true,
            error: "".to_string(),
        }))
    }

    async fn download_file(
        &self,
        request: Request<DownloadFileRequest>,
    ) -> ServiceResult<DownloadFileResponse> {
        let request = request.into_inner();
        info!(target: LOG_TARGET, "We got a new download file request: {:?}", request);
        // downloading metadata from any peers
        let (tx, rx) = oneshot::channel();
        self.p2p_command_sender
            .send(P2pCommand::RequestMetadata {
                request: MetadataDownloadRequest {
                    file_id: request.file_id,
                },
                result: tx,
            })
            .await
            .map_err(|e| Status::internal(format!("Failed to send p2p command: {}", e)))?;
        let metadata = rx
            .await
            .map_err(|e| Status::internal(format!("Failed to get file metadata: {}", e)))?;

        info!(target: LOG_TARGET, "file metadata: filename {} ({:?})", metadata.original_file_name, metadata.merkle_root);

        // save metadata file to local DB (into a new column family for downloading chunks)
        let download_path = PathBuf::from(request.download_path).join(format!(
            "{}_{}",
            metadata.hash_id(),
            metadata.original_file_name.clone().replace(".", "_") // 目录不能带"."
        ));

        let pending_download_record = PendingDownloadRecord::new(
            request.file_id,
            metadata.original_file_name.clone(),
            download_path,
        );
        pending_download_record
            .save_metadata(&metadata)
            .await
            .map_err(|e| Status::internal(format!("Failed to save metadata: {}", e)))?;
        self.file_store
            .add_published_file_record(metadata.into())
            .map_err(|e| Status::internal(format!("Failed to add published file record: {}", e)))?;
        self.file_store
            .add_pending_download(pending_download_record)
            .map_err(|e| {
                Status::internal(format!("Failed to add pending download record: {}", e))
            })?;

        info!(target: LOG_TARGET, "file {} downloaded successfully", request.file_id);

        Ok(Response::new(DownloadFileResponse {
            success: true,
            error: "".to_string(),
        }))
    }

    async fn get_public_files(
        &self,
        request: Request<GetPublicFilesRequest>,
    ) -> ServiceResult<Self::GetPublicFilesStream> {
        let request = request.into_inner();
        info!(target: LOG_TARGET, "We got a new get public files request: {:?}", request);

        let (tx, mut rx) = mpsc::channel::<PublicPublishedFile>(100);
        let (resp_tx, resp_rx) = mpsc::channel::<Result<GetPublicFilesResponse, Status>>(100);
        self.p2p_command_sender
            .send(P2pCommand::GetPublicFiles { result: tx })
            .await
            .map_err(|e| {
                Status::internal(format!(
                    "Failed to send p2p command to get public files available: {}",
                    e
                ))
            })?;
        tokio::spawn(async move {
            while let Some(result) = rx.recv().await {
                info!(target: LOG_TARGET, "get public files available: {:?}", &result);
                if let Err(e) = resp_tx
                    .send(Ok(GetPublicFilesResponse {
                        file_id: result.file_id,
                        original_file_name: result.original_file_name,
                    }))
                    .await
                {
                    error!(target: LOG_TARGET, "Failed to send get public files response: {:?}", e);
                };
            }
        });

        let stream = ReceiverStream::new(resp_rx);
        Ok(Response::new(Box::pin(stream)))
    }
}

#[derive(Debug)]
pub struct GrpcService<FileStore>
where
    FileStore: store::Store + Send + Sync + 'static,
{
    port: u16,
    file_publish_sender: mpsc::Sender<FileProcessResult>,
    p2p_command_sender: mpsc::Sender<P2pCommand>,
    file_store: FileStore,
}

impl<FileStore> GrpcService<FileStore>
where
    FileStore: store::Store + Send + Sync + 'static,
{
    pub fn new(
        port: u16,
        file_publish_sender: mpsc::Sender<FileProcessResult>,
        p2p_command_sender: mpsc::Sender<P2pCommand>,
        file_store: FileStore,
    ) -> Self {
        Self {
            port,
            file_publish_sender,
            p2p_command_sender,
            file_store,
        }
    }
}

#[async_trait::async_trait]
impl<FileStore> Service for GrpcService<FileStore>
where
    FileStore: store::Store + Send + Sync + Clone + 'static,
{
    async fn start(&mut self, cancel_token: CancellationToken) -> Result<(), ServerError> {
        let addr = format!("127.0.0.1:{}", self.port).parse()?;
        info!(target: LOG_TARGET, "gRPC server start on {}", addr);
        if let Err(e) = Server::builder()
            .add_service(DfsServer::new(DfsGrpcService::new(
                self.file_publish_sender.clone(),
                self.p2p_command_sender.clone(),
                self.file_store.clone(),
            )))
            .serve_with_shutdown(addr, cancel_token.cancelled())
            .await
        {
            error!(target: LOG_TARGET, "gRPC server start error: {:?}", e);
        };
        info!(target: LOG_TARGET, "gRPC server stopped");
        Ok(())
    }
}
